package com.titan.single.kafka;

import com.fasterxml.jackson.databind.ObjectMapper;
import com.titan.single.constant.KafkaConstant;
import com.titan.single.vo.KafkaMessage;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.kafka.support.KafkaHeaders;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.messaging.handler.annotation.Payload;
import org.springframework.stereotype.Service;

/**
 * 消费者
 */
@Service
@Slf4j
public class KafkaConsumer {

    @Autowired
    private ObjectMapper objectMapper;

    /**
     * 消费简单文本消息
     */
    @KafkaListener(topics = KafkaConstant.TOPIC_MSG, groupId = KafkaConstant.GROUP_TIAN)
    public void consumeMessage(@Payload String message,
                              @Header(KafkaHeaders.RECEIVED_TOPIC) String topic,
                              @Header(KafkaHeaders.RECEIVED_PARTITION) int partition,
                              @Header(KafkaHeaders.OFFSET) long offset,
                              Acknowledgment acknowledgment) {
        try {
            log.info("消费者---接收到消息: topic=[{}], partition=[{}], offset=[{}], message=[{}]",
                    topic, partition, offset, message);
            
            // 处理业务逻辑
            processMessage(message);
            
            // 手动提交偏移量
            acknowledgment.acknowledge();
            log.info("消费者---消息处理完成并提交偏移量");
            
        } catch (Exception e) {
            log.error("消费者---消费消息异常: {}", e.getMessage(), e);
            // 根据业务需要决定是否提交偏移量
        }
    }

    /**
     * 消费JSON格式消息
     */
    @KafkaListener(topics = KafkaConstant.TOPIC_JSON_MSG, groupId = KafkaConstant.GROUP_TIAN)
    public void consumeJsonMessage(@Payload String message,
                                  ConsumerRecord<String, String> record,
                                  Acknowledgment acknowledgment) {
        try {
            log.info("接收到JSON消息: key=[{}], value=[{}], topic=[{}], partition=[{}], offset=[{}]",
                    record.key(), record.value(), record.topic(), record.partition(), record.offset());
            
            // 解析JSON消息
            KafkaMessage kafkaMessage = objectMapper.readValue(message, KafkaMessage.class);
            log.info("解析后的消息对象: {}", kafkaMessage);
            
            // 处理业务逻辑
            processJsonMessage(kafkaMessage);
            
            // 手动提交偏移量
            acknowledgment.acknowledge();
            log.info("消费者---消息处理完成并提交偏移量");
            
        } catch (Exception e) {
            log.error("消费JSON消息异常: {}", e.getMessage(), e);
        }
    }

    /**
     * 消费带key的相关消息-分区路由机制
     * 这个key最主要的作用是**决定消息被发送到哪个分区（Partition）**：
     * - **有key的情况**：Kafka使用key的哈希值来决定消息应该发送到哪个分区
     * - **无key的情况**：消息会以轮询（Round-robin）的方式分发到各个分区
     */
    @KafkaListener(topics = KafkaConstant.TOPIC_KEY, groupId = KafkaConstant.GROUP_TIAN)
    public void consumeUserMessage(@Payload String message,
                                  @Header(KafkaHeaders.RECEIVED_KEY) String key,
                                  Acknowledgment acknowledgment) {
        try {
            log.info("接收到用户消息: key=[{}], message=[{}]", key, message);
            
            // 处理用户相关业务
            processUserMessage(key, message);
            
            acknowledgment.acknowledge();
            log.info("消费者---消息处理完成并提交偏移量");
            
        } catch (Exception e) {
            log.error("消费用户消息异常: {}", e.getMessage(), e);
        }
    }



    /**
     * 处理简单文本消息的业务逻辑
     */
    private void processMessage(String message) {
        // 在这里添加你的业务处理逻辑

        // 模拟业务处理
        try {
            Thread.sleep(100); // 模拟处理时间
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            log.error("消息处理被中断: {}", message);
        }
    }

    /**
     * 处理JSON消息的业务逻辑
     */
    private void processJsonMessage(KafkaMessage kafkaMessage) {
        // 添加你的JSON消息处理逻辑

    }

    /**
     * 处理用户消息的业务逻辑
     */
    private void processUserMessage(String key, String message) {
        // 添加你的用户消息处理逻辑

    }
}